

# 最大的特点就是消费并不需要确保提供方存在,实现了服务之间的高度解耦
#   可以用它来：解耦、异步、削峰。


# rabbitmq 的使用场景
#     （1）服务间异步通信
#     （2）顺序消费
#     （3）定时任务
#     （4）请求削峰


# 微服务架构中服务间通信（ 应用解耦） 
#         订单服务下单后，通过 RabbitMQ 通知库存服务扣减库存
#         用户服务注册后，通知邮件服务发送欢迎邮件


# 耗时操作异步化，提升系统响应速度（异步处理）
#         用户上传视频后立即返回成功，视频转码任务放入队列
#         支付成功后异步生成电子发票

# 应对突发高并发流量（流量削峰）
#         电商秒杀活动：将海量请求放入队列，有序处理订单
#         票务系统售票：控制数据库写入速率



# 跨服务数据一致性（分布式事务） 通过 最终一致性 替代强一致性 使用 RabbitMQ 的确认机制保证消息可靠传递

# 集中式日志处理（日志收集）多台服务器将日志发送到 RabbitMQ  避免日志写入影响主业务性能

# 事件广播（发布/订阅）  使用 fanout 或 topic 交换机  一条消息触发多个独立系统的动作
#         订单状态变更时：
#         ✓ 通知用户APP
#         ✓ 更新搜索引擎索引
#         ✓ 触发风控系统分析

# 跨服务同步调用（RPC（远程过程调用））


# # 定时/延迟执行任务（延迟任务）
#         订单30分钟未支付自动取消
#         预约提醒提前1小时通知



# # 多系统间数据同步（数据同步）
#         主数据库变更同步到缓存
#         核心系统数据同步到分析平台


# https://www.bilibili.com/video/BV1oCwEeVEe4/?spm_id_from=333.337.search-card.all.click&vd_source=55a70ceabdb66cf8fe463891b9b7b436

# 消息队列：   


# 生产者--------------->  队列  ---------------------->消费者
# 当消息处理不过来时， 暂存在队列慢慢处理



# 不同的消息 在不同的队列中，不同的队列时不同的进程


# exchange--> 消息分发路由，将消息发到不同的队列

# 维护在exchange中的 路由方式 和绑定关系，----->  元素据

# 每个服务器上的 rabbitMQ 示例 -----------> broker 

# 在这个队列的基础上实现了： 延时队列、死信队列、优先级队列

# 优先级队列， 区分vip 和非vip，消费者优先消费 优先级高的----------->  当服务器攻击时 免费用户很卡，但是 付费用户依然很好


# 普通集群： 每个都是相同的个体，支持读写， 互相同步  exchange  中的元数据  不会同步 queue 数据
# 如果访问  q1 的数据， q1有 那么直接返回，如果q1 没有，那么通过 exchange 找到 他在那个队列，从相应队列中拿到数据 返回
# 通过 增加 broker 增强服务， 但是互相不同步 queue 数据，如果一个broker 挂了，会导致这个 队列不可用




# 在普通集群基础上： 增加queue 备份（镜像队列）


# quorum队列集群： 使用 raft（更靠谱的一致性算法） 同步多个 broker 的元数据和 队列数据；  引入选举机制 解决分区问题




# 生产者------------->  channel----------- broker ----------------- channel -----------消费者

# 多个 channel  组成 connection 
# 多个  virtual host  组成 broker 
# 多个 exchange 组成  vhost  
# 一个 exchange 包含多个 queue

# 经过一次 connection 链接  可以发送多个消息，每个消息发送占用一个  channel 



# Celery + Redis/RabbitMQ​​
# ​​功能​​：定时任务、工作流编排、结果回溯
# ​​优势​​：支持百万级任务队列，flower组件提供监控UI

from review.celery_demo import Celery
app = Celery('tasks', broker='redis://localhost/0')
@app.task
def process_data(data):
    return data.upper()



# ---------------------exchange 类型-------------------
# direct (直连交换机)
# 路由规则： 精确匹配 routing key  只有完全匹配成功的队列才会收到该消息。  
# 典型应用场景： 点对点精确路由、基于不同业务类型或严重级别分发消息
# 类比： 就像寄信，信封上写的是 "技术部-张三"，只有 "技术部-张三" 这个信箱会收到。
import pika
import sys

# 生产者
def direct_producer():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    # 声明 direct 类型交换机
    channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
    
    severity = sys.argv[1] if len(sys.argv) > 1 else 'info'
    message = ' '.join(sys.argv[2:]) or "Hello Direct Exchange!"
    
    # 发送消息时指定 routing_key
    channel.basic_publish(
        exchange='direct_logs',
        routing_key=severity,  # 精确匹配的键
        body=message
    )
    print(f" [x] Sent {severity}:{message}")
    connection.close()

# 消费者
def direct_consumer(severities):
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    channel.exchange_declare(exchange='direct_logs', exchange_type='direct')
    result = channel.queue_declare(queue='', exclusive=True)
    queue_name = result.method.queue
    
    # 绑定多个 routing_key
    for severity in severities:
        channel.queue_bind(
            exchange='direct_logs',
            queue=queue_name,
            routing_key=severity
        )
    
    print(' [*] Waiting for logs. To exit press CTRL+C')
    
    def callback(ch, method, properties, body):
        print(f" [x] {method.routing_key}:{body.decode()}")
    
    channel.basic_consume(
        queue=queue_name,
        on_message_callback=callback,
        auto_ack=True
    )
    channel.start_consuming()

# 使用示例：
# 生产者: python script.py error "Database connection failed"
# 消费者: direct_consumer(['error', 'warning'])






# fanout (扇出交换机 / 广播交换机)
# 路由规则： 忽略 routing key，进行广播。
# 生产者发送消息时指定的 routing key 会被 fanout 交换机忽略。
# 交换机会将收到的每一条消息都复制一份，并发送到所有绑定到该交换机上的队列。
# 典型应用场景： 广播通知、发布/订阅模式（例如，一个新闻发布事件需要同时更新用户通知、生成缓存、记录日志等多个独立系统）
# 类比： 就像大喇叭广播，所有连接到这个喇叭的听众（队列）都能听到同样的消息。
import pika

# 生产者
def fanout_producer():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    # 声明 fanout 类型交换机
    channel.exchange_declare(exchange='fanout_logs', exchange_type='fanout')
    
    message = "Broadcast message to all queues!"
    channel.basic_publish(
        exchange='fanout_logs',
        routing_key='',  # 被忽略
        body=message
    )
    print(f" [x] Sent {message}")
    connection.close()

# 消费者
def fanout_consumer(consumer_name):
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    channel.exchange_declare(exchange='fanout_logs', exchange_type='fanout')
    result = channel.queue_declare(queue='', exclusive=True)
    queue_name = result.method.queue
    
    # 绑定时不需指定 routing_key
    channel.queue_bind(exchange='fanout_logs', queue=queue_name)
    
    print(f' [{consumer_name}] Waiting for broadcasts')
    
    def callback(ch, method, properties, body):
        print(f" [{consumer_name}] Received {body.decode()}")
    
    channel.basic_consume(
        queue=queue_name,
        on_message_callback=callback,
        auto_ack=True
    )
    channel.start_consuming()

# 使用示例：
# 启动多个消费者: fanout_consumer("Consumer1"), fanout_consumer("Consumer2")
# 生产者: fanout_producer()






# topic (主题交换机)
# 路由规则： 基于 routing key 的模式匹配（通配符）。
# 队列在绑定到交换机时需要指定一个 binding key。这个 binding key 可以包含特殊通配符：
#         * (星号)：匹配一个单词（由 . 分隔的部分）。例如 *.orange.*。
#         # (井号)：匹配零个或多个单词。例如 agreement.#。
# 特点： topic 交换机功能最强大、最灵活，可以实现 direct 和 fanout 的大部分功能。
# 典型应用场景： 基于多种属性（如来源、类型、动作、严重性等组合）进行灵活的消息路由（例如，路由所有 *.error 消息到错误中心，路由所有 agreement.eu.berlin.# 消息到柏林欧盟协议处理服务）。
# 类比： 就像订阅杂志，你可以订阅 "科技.*"（所有科技类杂志），或者 "摄影.技巧.#"（所有摄影技巧相关杂志，无论初级还是高级）。杂志社（交换机）根据杂志类别（routing key）和你的订阅规则（binding key）决定是否给你寄送。
import pika
import sys

# 生产者
def topic_producer():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    # 声明 topic 类型交换机
    channel.exchange_declare(exchange='topic_logs', exchange_type='topic')
    
    routing_key = sys.argv[1] if len(sys.argv) > 1 else 'anonymous.info'
    message = ' '.join(sys.argv[2:]) or "Hello Topic Exchange!"
    
    channel.basic_publish(
        exchange='topic_logs',
        routing_key=routing_key,
        body=message
    )
    print(f" [x] Sent {routing_key}:{message}")
    connection.close()

# 消费者
def topic_consumer(binding_keys):
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    channel.exchange_declare(exchange='topic_logs', exchange_type='topic')
    result = channel.queue_declare(queue='', exclusive=True)
    queue_name = result.method.queue
    
    # 绑定模式键
    for binding_key in binding_keys:
        channel.queue_bind(
            exchange='topic_logs',
            queue=queue_name,
            routing_key=binding_key
        )
    
    print(f' [*] Waiting for topics: {binding_keys}')
    
    def callback(ch, method, properties, body):
        print(f" [x] {method.routing_key}:{body.decode()}")
    
    channel.basic_consume(
        queue=queue_name,
        on_message_callback=callback,
        auto_ack=True
    )
    channel.start_consuming()

# 使用示例：
# 生产者: python script.py "order.eu.berlin.paid"
# 消费者1: topic_consumer(["*.error"])        # 接收所有 error
# 消费者2: topic_consumer(["order.#"])        # 接收所有订单消息
# 消费者3: topic_consumer(["*.berlin.*"])     # 接收柏林相关消息







# headers (头交换机)
# 路由规则： 基于消息 headers 属性（键值对）进行匹配，忽略 routing key。
# 工作方式：
# 队列在绑定到交换机时，需要指定一组 headers 键值对作为匹配条件。
# 生产者发送消息时，可以在消息的 headers 属性中设置任意的键值对。
# 交换机在路由时，会检查消息的 headers 是否完全匹配绑定时指定的条件（默认是 "all" 匹配，即所有指定的键值对都必须匹配且值相等；也可以设置为 "any" 匹配，即匹配任意一个指定的键值对）。
# 匹配成功的队列会收到消息。
# 特点： 路由规则不依赖字符串，而是基于键值对，提供了另一种维度的灵活性，但性能通常不如基于 routing key 的匹配高效，使用相对较少。
# 典型应用场景： 当路由决策需要基于多个非结构化属性（如协议版本、内容类型、自定义标志等）时。
# 类比： 就像海关检查，根据护照上的各种签证和印章（headers）来决定是否允许入境（路由到队列），而不看目的地国家（routing key）。
import pika
import json

# 生产者
def headers_producer():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    # 声明 headers 类型交换机
    channel.exchange_declare(exchange='headers_logs', exchange_type='headers')
    
    message = "Message with special headers"
    headers = {
        'type': 'report',
        'format': 'pdf',
        'x-match': 'all'  # 可改为 'any' 测试不同匹配模式
    }
    
    properties = pika.BasicProperties(headers=headers)
    
    channel.basic_publish(
        exchange='headers_logs',
        routing_key='',  # 被忽略
        body=message,
        properties=properties
    )
    print(f" [x] Sent headers:{json.dumps(headers)} message:{message}")
    connection.close()

# 消费者
def headers_consumer(bind_headers, match_type='all'):
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    
    channel.exchange_declare(exchange='headers_logs', exchange_type='headers')
    result = channel.queue_declare(queue='', exclusive=True)
    queue_name = result.method.queue
    
    # 绑定头信息
    channel.queue_bind(
        exchange='headers_logs',
        queue=queue_name,
        arguments={**bind_headers, 'x-match': match_type}
    )
    
    print(f' [*] Waiting for headers (match={match_type}): {bind_headers}')
    
    def callback(ch, method, properties, body):
        print(f" [x] Received headers:{properties.headers} message:{body.decode()}")
    
    channel.basic_consume(
        queue=queue_name,
        on_message_callback=callback,
        auto_ack=True
    )
    channel.start_consuming()

# 使用示例：
# 生产者: headers_producer()
# 消费者1: headers_consumer({'type': 'report', 'format': 'pdf'})  # 全匹配
# 消费者2: headers_consumer({'type': 'alert'}, 'any')            # 任意匹配








# 为什么使用MQ
# 异步处理 - 相比于传统的串行、并行方式，提高了系统吞吐量。
# 应用解耦 - 系统间通过消息通信，不用关心其他系统的处理。
# 流量削锋 - 可以通过消息队列长度控制请求量；可以缓解短时间内的高并发请求。
# 日志处理 - 解决大量日志传输。
# 消息通讯 - 消息队列一般都内置了高效的通信机制，因此也可以用在纯的消息通讯。比如实现点对点消息队列，或者聊天室等




# MQ的优点
# 主要是：解耦、异步、削峰。
# 解耦：A 系统发送数据到 BCD 三个系统，通过接口调用发送。如果 E 系统也要这个数据呢？那如果 C 系统现在不需要了呢？
#       A 系统负责人几乎崩溃…A 系统跟其它各种乱七八糟的系统严重耦合，A 系统产生一条比较关键的数据，
#       很多系统都需要 A 系统将这个数据发送过来。如果使用 MQ，A 系统产生一条数据，发送到 MQ 里面去，
#       哪个系统需要数据自己去 MQ 里面消费。如果新系统需要数据，直接从 MQ 里消费即可；
#       如果某个系统不需要这条数据了，就取消对 MQ 消息的消费即可。这样下来，A 系统压根儿不需要去考虑要给谁发送数据，
#       不需要维护这个代码，也不需要考虑人家是否调用成功、失败超时等情况。
#     -- 就是一个系统或者一个模块，调用了多个系统或者模块，互相之间的调用很复杂，维护起来很麻烦。但是其实这个调用是不需要直接同步调用接口的，如果用 MQ 给它异步化解耦。

# 异步：A 系统接收一个请求，需要在自己本地写库，还需要在 BCD 三个系统写库，自己本地写库要 3ms，BCD 三个系统分别写库要 300ms、450ms、200ms。
#       最终请求总延时是 3 + 300 + 450 + 200 = 953ms，接近 1s，用户感觉搞个什么东西，慢死了慢死了。用户通过浏览器发起请求。
#       如果使用 MQ，那么 A 系统连续发送 3 条消息到 MQ 队列中，假如耗时 5ms，A 系统从接受一个请求到返回响应给用户，总时长是 3 + 5 = 8ms。


# 削峰：减少高峰时期对服务器压力。主要是还是来自于互联网的业务场景，例如，马上即将开始的春节火车票抢购，大量的用户需要同一时间去抢购；
#       以及大家熟知的阿里双11秒杀，这个时候在同一时间点请求来的太猛烈，服务器受不住。削峰，等于是减缓服务器压力，可以以漏斗，
#       或者水管的方式来进行消费，想一想水管和漏斗的机制大家就明白了




# MQ 的常见问题有：
#         消息的顺序问题
#         消息的重复问题



# RabbitMQ基本概念
# Broker： 简单来说就是消息队列服务器实体
# Exchange： 消息交换机，它指定消息按什么规则，路由到哪个队列
# Queue： 消息队列载体，每个消息都会被投入到一个或多个队列
# Binding： 绑定，它的作用就是把exchange和queue按照路由规则绑定起来
# Routing Key： 路由关键字，exchange根据这个关键字进行消息投递
# VHost： vhost 可以理解为虚拟 broker ，即 mini-RabbitMQ server。其内部均含有独立的 queue、exchange 和 binding 等，但最最重要的是，其拥有独立的权限系统，可以做到 vhost 范围的用户控制。当然，从 RabbitMQ 的全局角度，vhost 可以作为不同权限隔离的手段（一个典型的例子就是不同的应用可以跑在不同的 vhost 中）。
# Producer： 消息生产者，就是投递消息的程序
# Consumer： 消息消费者，就是接受消息的程序
# Channel： 消息通道，在客户端的每个连接里，可建立多个channel，每个channel代表一个会话任务
# 由Exchange、Queue、RoutingKey三个才能决定一个从Exchange到Queue的唯一的线路。









# 直播弹幕与实时互动（选RabbitMQ）
# 需求：低延迟传递弹幕、礼物消息（强实时性）。
# 优势：
#     RabbitMQ内存队列+直接交换器路由，延迟更低510。
#     支持优先级队列（VIP弹幕优先处理）1。
#     Python示例（pika）：



# 视频转码任务调度（选RabbitMQ）
# 需求：异步分发转码任务，需可靠执行与重试。
# 优势：
# RabbitMQ死信队列+ACK机制实现任务重试14。
# 结合Celery（Python任务队列框架）更便捷6。



# 用户请求 → Kafka（日志/行为） → 分析系统
# 直播流 → RabbitMQ（弹幕/礼物） → 实时推送
# 视频上传 → RabbitMQ（转码任务） → 转码集群 → Kafka（转码完成事件）


# Kafka主导：用于数据管道（日志、指标流），发挥高吞吐与流处理优势。
# RabbitMQ主导：用于业务操作（弹幕、任务调度），利用低延迟与灵活路由。




# -----------------------------------kafka--------------------------------------------------------

# 如果是大数据领域的实时计算、日志采集等场景，用 Kafka 是业内标准的，绝对没问题，社区活跃度很高，绝对不会黄，何况几乎是全世界这个领域的事实性规范


